package com.uestc.coprocessor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.uestc.coprocessor.JAggregateProtocol.CountRequest;
import com.uestc.coprocessor.JAggregateProtocol.CountResponse;

public class MultiColumnAggregation extends JAggregateProtocol.RowCountService
		implements Coprocessor, CoprocessorService {
	public static final String COLUMNNAME_QINIU = "Qiniu";
	public static final String COLUMNNAME_JINSHAN = "Jinshan";
	public static final String COLUMNNAME_QINIU2 = "Qiniu2";
	private RegionCoprocessorEnvironment env;

	@Override
	public void start(CoprocessorEnvironment arg0) throws IOException {
		if (arg0 instanceof RegionCoprocessorEnvironment) {
			this.env = (RegionCoprocessorEnvironment) arg0;
		} else {
			throw new CoprocessorException("Must be loaded on a table region!");
		}
	}

	@Override
	public void stop(CoprocessorEnvironment arg0) throws IOException {

	}

	@Override
	public Service getService() {
		return this;
	}

	@Override
	public void getSizes(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
		long[] values = { 0, 0, 0, 0 };
		String columns = request.getColumns();
		if (columns == null || "".equals(columns))
			throw new NullPointerException("you need specify the columns");
		String[] columnArray = columns.split(";");

		// 设置filter，只过滤出我们需要的列
		Filter filter = null;
		ArrayList<Filter> filters = new ArrayList<>();
		for (String column : columnArray) {
			filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("location"),
					CompareFilter.CompareOp.EQUAL, Bytes.toBytes(column));
			filters.add(filter);
		}
		filter = new FilterList(FilterList.Operator.MUST_PASS_ONE, filters);
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("location"));
		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("chunksize"));
		scan.setFilter(filter);

		JAggregateProtocol.CountResponse response = null;
		InternalScanner scanner = null;
		try {
			scanner = env.getRegion().getScanner(scan);
			List<Cell> results = new ArrayList<Cell>();
			boolean hasMore = false;
			do {
				hasMore = scanner.next(results);
				if (results.size() < 2)
					continue;
				// chunksize 在前，location在后
				Cell kv0 = results.get(0);
				long chunksize = Long.parseLong(Bytes.toString(CellUtil.cloneValue(kv0)));
				Cell kv1 = results.get(1);
				String location = Bytes.toString(CellUtil.cloneValue(kv1));
				switch (location) {
				case COLUMNNAME_QINIU:
					values[0] += chunksize;
					break;
				case COLUMNNAME_QINIU2:
					values[1] += chunksize;
					break;
				case COLUMNNAME_JINSHAN:
					values[2] += chunksize;
					break;
				default:
					break;
				}
				results.clear();
			} while (hasMore);

			// 生成response
			response = JAggregateProtocol.CountResponse.newBuilder().setQiniuCount(values[0]).setQiniu2Count(values[1])
					.setJinshanCount(values[2]).build();

		} catch (IOException e) {
			e.printStackTrace();
			ResponseConverter.setControllerException(controller, e);
		} finally {
			if (scanner != null) {
				try {
					scanner.close();
				} catch (IOException ignored) {
				}
			}
		}
		done.run(response);
	}

}
